package com.simple.datax;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;
import java.util.UUID;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.api.Connector;
import com.simple.datax.api.ConnectorException;
import com.simple.datax.conn.BaseConnector;
import com.simple.datax.conn.impl.ConnectorFactory;

/**
 * @Title: EPSimulator.java
 * @Package com.simple.datax
 * @Description: TODO(添加描述)
 * @author Paladin Chen
 * @date 2013-2-5 下午3:29:13
 * @version V1.0
 */
public class EPSimulator {
	private static final Logger logger = LoggerFactory.getLogger(EPSimulator.class);

	private int speed;
	private String encoding = "utf-8";

	public static void main(String args[]) {
		System.getProperties().list(System.out);

		if (args.length != 1) {
			logger.error("[Usage] java LNSDataSimulator Operation:");
			logger.error(" Operation option:");
			logger.error(" 1: File to Ftp.");
			logger.error(" 2: File to Ftp and Ftp to file 1.");
			logger.error(" 3: File to Ftp and Ftp to file 1 and Ftp to file 2.");
			return;
		}

		String operation = args[0];
		if (!operation.equals("1") && !operation.equals("2") && !operation.equals("3")) {
			logger.error("Command parameter error. Input 1 or 2 or 3.");
			return;
		}

		Properties prop = new Properties();
		FileInputStream fis = null;
		String scanFilePath = null;
		String scanFtp = null;
		String ftpRecv = null;
		String recvPath = null;
		String ftpRecv2 = null;
		String recvPath2 = null;
		String speed = null;

		try {
			fis = new FileInputStream("ep.properties");
			prop.load(fis);

			scanFilePath = prop.getProperty("scan_filepath");
			if (scanFilePath == null || scanFilePath.trim().isEmpty()) {
				logger.error("Please set message template file path.");
				return;
			}

			speed = prop.getProperty("simulator_speed");
			if (speed == null || speed.trim().isEmpty()) {
				logger.error("Please set speed.");
				return;
			}

			scanFtp = prop.getProperty("ftp_send");
			if (scanFtp == null || scanFtp.trim().isEmpty()) {
				logger.error("Please set ftp_send.");
				return;
			}

			ftpRecv = prop.getProperty("ftp_recv1");
			if (ftpRecv == null || ftpRecv.trim().isEmpty()) {
				logger.error("Please set ftp_recv1.");
				ftpRecv = null;
			}

			recvPath = prop.getProperty("recv1_filepath");
			if (recvPath == null || recvPath.trim().isEmpty()) {
				logger.error("Please set recv1_filepath.");
				recvPath = null;
			}

			ftpRecv2 = prop.getProperty("ftp_recv2");
			if (ftpRecv2 == null || ftpRecv2.trim().isEmpty()) {
				logger.error("Please set ftp_recv2.");
				ftpRecv2 = null;
			}

			recvPath2 = prop.getProperty("recv2_filepath");
			if (recvPath2 == null || recvPath2.trim().isEmpty()) {
				logger.error("Please set recv2_filepath.");
				recvPath2 = null;
			}

		} catch (Exception ex) {
			logger.error(ex);
			return;
		} finally {
			try {
				if (fis != null) {
					fis.close();
				}
			} catch (IOException e) {
				logger.error(e);
			}
		}

		EPSimulator eps = new EPSimulator(Integer.valueOf(speed));
		if (operation.equals("1")) {
			eps.startProcThread("FILE", scanFilePath, "FTP", scanFtp);
		} else if (operation.equals("2")) {
			eps.startProcThread("FILE", scanFilePath, "FTP", scanFtp);
			if(ftpRecv != null && recvPath != null){
				eps.startProcThread("FTP", ftpRecv, "FILE", recvPath);
			}
		} else if (operation.equals("3")) {
			eps.startProcThread("FILE", scanFilePath, "FTP", scanFtp);
			if(ftpRecv != null && recvPath != null){
				eps.startProcThread("FTP", ftpRecv, "FILE", recvPath);
			}
			if(ftpRecv2 != null && recvPath2 != null){
				eps.startProcThread("FTP", ftpRecv2, "FILE", recvPath2);
			}
		}
	}

	public EPSimulator(int speed) {
		super();
		this.speed = speed;
	}

	/**
	 * 开始接受结果数据
	 * 
	 * @param connector
	 */
	public void startProcThread(String recvType, String recvPara, String sendType, String sendPara) {
		Thread thread = new Thread(new DataRecvAndSend(initConnector(recvType, recvPara), initConnector(sendType,
				sendPara)));
		thread.start();
	}

	/**
	 * 初始化所有指定连接器
	 * 
	 * @param connector
	 */
	private BaseConnector initConnector(String type, String param) {
		if (type == null || param == null) {
			return null;
		}
		String connPara = param;
		connPara = type + ConnectorFactory.KEY_SPLITTER + connPara;
		try {
			BaseConnector conn = ConnectorFactory.getInstance().makeObject(connPara);
			conn.setEncoding(encoding);
			logger.debug("Return Connector:"+ connPara);
			return conn;
		} catch (Exception e) {
			logger.error("Fail to initial connector - " + connPara, e);
		}

		return null;
	}

	/**
	 * 模拟数据，并发送至指定连接器
	 * 
	 * @author Jin Yi
	 * 
	 */
	public class DataRecvAndSend implements Runnable {

		private Connector recvConn;
		private Connector sendConn;

		public DataRecvAndSend(BaseConnector recvConn, BaseConnector sendConn) {
			this.recvConn = recvConn;
			this.sendConn = sendConn;
		}

		@Override
		public void run() {
			while (true) {
				logger.info("running...");
				getMessage();
				try {
					Thread.sleep(speed * 1000);
				} catch (Exception ex) {

				}
			}

		}

		private void sendMessage(String message) {
			SimpleMessage sm = new SimpleMessage();
			sm.setContent(message);
			sm.setRecvRawMsgid(UUID.randomUUID().toString());
			sendConn.send(sm);
			// logger.debug(message);
			logger.info("======== Total Send ");
		}

		private void getMessage() {
			String str = null;

			try {
				str = recvConn.getMessage();
				while (str != null) {
					logger.debug("Get Msg from:"+recvConn.toString());
					sendMessage(str);
					str = recvConn.getMessage();
				}
			} catch (ConnectorException e) {
				logger.error("Failed to get message.", e);
			}

		}

	}
}
